package org.bikelab.se3.job;

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.bikelab.se3.rdf.reader.NoLineToParseException;
import org.bikelab.se3.rdf.reader.RdfReaderFactory;
import org.bikelab.se3.rdf.reader.WrongNumberOfSegmentsException;
import org.semanticweb.yars.nx.Node;


public class PropertyCountJob {
	public static class SortMapper  extends Mapper<Object, Text, IntWritable,Text>{
		private Text property = new Text();
	    private IntWritable count= new IntWritable();
	    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
	    	//System.out.println("["+value.toString()+"]");
	    	  property.set(value.toString().substring(0,value.toString().indexOf('>')+1));
	    	  StringTokenizer itr = new StringTokenizer(value.toString());
	    	  for(int i=0;i<itr.countTokens()-1;i++){
	    		  itr.nextToken();
	    	  }try{
	    	  count.set(Integer.parseInt(itr.nextToken()));
	        context.write(count,property);
	    	  }catch(Exception e){
	    		  System.err.println(e.getMessage());
	    	  }
	    }
	  }
	
	  public static class ParserMapper   extends Mapper<Object, Text, Text, IntWritable>{
		    private final static IntWritable one = new IntWritable(1);
		    private Text property = new Text();
		      
		    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
			try {
			   Node[] quad;
				quad = RdfReaderFactory.parseNqLine(value.toString());
		        property.set(quad[1].toN3());
		        context.write(property, one);
			} catch (WrongNumberOfSegmentsException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (NoLineToParseException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		    }
		  }
	  //
	  public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> {
		  //
        private IntWritable result = new IntWritable();
        public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
          int sum = 0;
          for (IntWritable val : values) {
            sum += val.get();
          }
          result.set(sum);
          context.write(key,result);
        }
      }
	  
	  public static void main(String[] args) throws Exception {
		/* String temp ="<http://www.w3.org/2000/01/rdf-schema#seeAlso>  1";
		  System.out.println(temp.toString().substring(0,temp.toString().indexOf('>')+1));
		  StringTokenizer itr = new StringTokenizer(temp.toString());
    	  for(int i=0;i<itr.countTokens()-1;i++){
    		  itr.nextToken();
    	  }
    	  
    	  System.out.println(itr.nextToken());*/
    	  
		  Configuration conf = new Configuration();
		  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		  if (otherArgs.length != 2) {
			  System.err.println("Usage: wordcount <in> <out>");
			  System.exit(2);
		  }
		  
		  Job job = new Job(conf, "word count");
		  job.setJarByClass(PropertyCountJob.class);
		  job.setMapperClass(ParserMapper.class);
		  job.setCombinerClass(IntSumReducer.class);
		  job.setReducerClass(IntSumReducer.class);
		  job.setOutputKeyClass(Text.class);
		  job.setOutputValueClass(IntWritable.class);
		  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		  if(job.waitForCompletion(true)){
			System.out.println("succefully counted properties.");  
		  }
		  
		  
		   job = new Job(conf, "sort");
		  job.setJarByClass(PropertyCountJob.class);
		  job.setMapperClass(SortMapper.class);
		  job.setOutputKeyClass(IntWritable.class);
		  job.setOutputValueClass(Text.class);
		  FileInputFormat.addInputPath(job, new Path(otherArgs[1]));
		  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]+"Sorted"));
		  
		  System.exit(job.waitForCompletion(true) ? 0 : 1);
		 }
}